package com.systar.activity;

import java.io.IOException;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.Watcher.Event.KeeperState;

/**
 * base class for zookeeper connection
 */
public abstract class ZookeeperConnector implements IActivable, Watcher
{
	private static final int TIMEOUT = 3000;
	
	private final String zkHostPort;
	private ZooKeeper zooKeeper;
	private AtomicBoolean isConnected;

	public ZookeeperConnector(String zkHostPort)
	{
		this.zkHostPort = zkHostPort;
		isConnected = new AtomicBoolean(false);
	}
	
	protected ZooKeeper getZooKeeper()
	{
		return zooKeeper;
	}

	@Override
	public boolean isActive()
	{
		return (zooKeeper != null) && isConnected.get();
	}
	
	@Override
	public void activate()
	{
		try
		{
			// establish connection to server
			connect();
			// initialization
			if (isActive())
			{
				startupInit();
			}
		}
		catch (Exception e)
		{
			e.printStackTrace();
			zooKeeper = null;
		}
	}

	protected abstract void startupInit() throws Exception;
	
	@Override
	public void deactivate()
	{
		disconnect();
	}

	private void connect() throws IOException
	{
		isConnected.set(false);
		synchronized (isConnected)
		{
			System.out.println("Starting Zookeeper");
			zooKeeper = new ZooKeeper(zkHostPort, TIMEOUT, this);
			try
			{
				isConnected.wait(TIMEOUT * 2);
			}
			catch (InterruptedException e)
			{
				// skip
				System.err.println(e.getMessage());
			}
		}
		if (!isConnected.get())
		{
			disconnect();
			System.out.println("Failed to start Zookeeper");
		}
		else
		{
			System.out.println("Zookeeper successfully started");
		}
	}

	@Override
	public void process(WatchedEvent event)
	{
		System.out.println("Zookeeper callback process");
		
		if (!isConnected.get())
		{
			// connection callback
			isConnected.set(event.getState() == KeeperState.SyncConnected);
			isConnected.notify();
		}
	}

	private void disconnect()
	{
		if (zooKeeper != null)
		{
			try
			{
				zooKeeper.close();
			}
			catch (InterruptedException e)
			{
				e.printStackTrace();
			}
			zooKeeper = null;
		}
	}
	
}
